package com.xuzimian.globaldemo.mq.rabbitmq.util;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.GetResponse;

import java.io.IOException;
import java.util.Map;
import java.util.Random;

/***
 * 队列分片包装器，
 * @author 徐自勉
 * 来源于《RabbitMQ实战指南》第九章
 */
public class RmqEncapsulation {


    /**
     * 分片数，表示一个逻辑队列背后的实际队列数。
     */
    private int subdivisionNum;

    public RmqEncapsulation(int subdivisionNum) {
        this.subdivisionNum = subdivisionNum;
    }

    /**
     * 声明队列
     *
     * @param channel
     * @param queue
     * @param durable
     * @param exclusive
     * @param autoDelte
     * @param arguments
     * @throws IOException
     */
    public void queueDelare(Channel channel, String queue, boolean durable,
                            boolean exclusive, boolean autoDelte, Map<String, Object> arguments) throws IOException {
        for (int i = 0; i < subdivisionNum; i++) {
            String queueName = queue + "_" + i;
            channel.queueDeclare(queueName, durable, exclusive, autoDelte, arguments);
        }
    }

    /**
     * 创建绑定关系
     *
     * @param channel
     * @param queue
     * @param routingKey
     * @param exchange
     * @param arguments
     * @throws IOException
     */
    public void queueDelare(Channel channel, String queue, String routingKey,
                            String exchange, Map<String, Object> arguments) throws IOException {
        for (int i = 0; i < subdivisionNum; i++) {
            String rkName = routingKey + "_" + i;
            String queueName = queue + "_" + i;
            channel.queueBind(queueName, exchange, rkName, arguments);
        }
    }


    /**
     * 随机挑选一个队列发送
     *
     * @param channel
     * @param exchange
     * @param routingKey
     * @param mandatory
     * @param props
     * @param body
     */
    public void basicPublis(Channel channel, String exchange, String routingKey,
                            boolean mandatory, AMQP.BasicProperties props, byte[] body) throws IOException {
        Random random = new Random();
        int index = random.nextInt(subdivisionNum);
        String rkName = routingKey + "_" + index;
        channel.basicPublish(exchange, rkName, mandatory, props, body);
    }

    /**
     * 拉模式封装
     * @param channel
     * @param queue
     * @param autoAck
     * @return
     */
    public GetResponse basicGet(Channel channel, String queue, boolean autoAck) throws IOException {
        GetResponse getResponse = null;
        Random random = new Random();
        int index = random.nextInt(subdivisionNum);
        String queueName=queue + "_" + index;
        getResponse= channel.basicGet(queueName, autoAck);
        if(null == getResponse){
            for(int i=0;i<subdivisionNum;i++){
                queueName=queue + "_" + i;
                getResponse= channel.basicGet(queueName, autoAck);
                if(null != getResponse){
                    return getResponse;
                }
            }
        }
        return getResponse;
    }


}
